package com.heima.safe.aliyun.mq;

import com.alibaba.fastjson.JSON;
import com.heima.safe.aliyun.verify.VerifyImage;
import com.heima.safe.aliyun.verify.VerifyText;
import com.sun.deploy.security.ruleset.ExceptionRule;
import org.apache.ibatis.javassist.bytecode.analysis.Executor;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.protocol.types.Field;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

import java.lang.reflect.Executable;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Consumer;

/**
 * @author mgw
 * @create 2021/10/6 20:38
 */
@Component
public class VerifyQueueListener {
    @Autowired
    private VerifyImage verifyImage;
    @Autowired
    private VerifyText verifyText;
    @Autowired
    private KafkaTemplate kafkaTemplate;
    //线程池
    private static ExecutorService executorService= Executors.newFixedThreadPool(10);
    @KafkaListener(topics = "VerifyQueue",groupId = "verifygroup")
    public void verifyQueue(ConsumerRecord<?,?> record, Acknowledgment acknowledgment) throws ExecutionException, InterruptedException {
        Map<String,Object> dateMap = JSON.parseObject(record.value().toString(), Map.class);
        Map<String,Object> resultMap=new HashMap<>();
        Future<List<Map<String, String>>> txtFuture = executorService.submit(new TextFutureResp(dateMap, verifyText));
        Future<List<Map<String, String>>> imageFuture = executorService.submit(new ImagesFutureResp(dateMap, verifyImage));
        List<Map<String, String>> txtResult = txtFuture.get();
        List<Map<String, String>> imageResult = imageFuture.get();
        ArrayList<Map<String, String>> allResult = new ArrayList<>();
        allResult.addAll(txtResult);
        allResult.addAll(imageResult);
        resultMap.put("verifyResult",allResult);
        resultMap.put("attach",dateMap.get("attach"));
        String backQueue = dateMap.get("backQueue").toString();
        kafkaTemplate.send(backQueue,JSON.toJSONString(resultMap));


    }
}
